package com.ot.flink.cdc.util;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * flink 运行环境统一初始化
 * @author : pangtaitao
 * @date : 2022/6/22
 */
public class EnvUtil {

    /**
     * 创建flink运行环境
     * @author : pangtaitao
     * @date : 2022/6/15
     */
    public static StreamExecutionEnvironment initEnv(StreamExecutionEnvironment flinkEnv) {
        //1.只有开启了CheckPointing,才会有重启策略
        flinkEnv.enableCheckpointing(300_000, CheckpointingMode.AT_LEAST_ONCE);
        flinkEnv.setRestartStrategy(RestartStrategies.fixedDelayRestart(12, Time.minutes(3)));
        // 方式快照过大
        flinkEnv.getCheckpointConfig().setMinPauseBetweenCheckpoints(1000);
        flinkEnv.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION) ;
        flinkEnv.setParallelism(3);
        return flinkEnv;
    }

}
